package com.cqk.bigdata.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

/**
 * @author: cqk
 * @date: 2018/4/2 14:06
 * @param
*/
public class ConsumerTest {
    public static void main(String[] args) {
        Properties kafkaProp = new Properties();
        kafkaProp.put("bootstrap.servers","spark001:9092");
        kafkaProp.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProp.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProp.put("group.id","consumer1");
        kafkaProp.put("enable.auto.commit","false");
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(kafkaProp);
        consumer.subscribe(Collections.singleton("kafka_test"));

        try{
            while (true){
                ConsumerRecords<String,String> records = consumer.poll(100);

                for ( ConsumerRecord<String,String> record:records) {
                    System.out.println(record.toString());

                    System.out.println("主题："+record.topic()+" 值："+record.value()+" 偏移量："+record.offset()
                            +" 分区：" +record.partition()+" 健"+record.key());
                }

                    // consumer.commitSync();//处理完当前批次消息，在轮询跟多消息之前，提交当前最新偏移量
                     consumer.commitAsync();//提交最后一个偏移量，让然后做其他的事，异步提交
    //            consumer.commitAsync(new OffsetCommitCallback() {
    //                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
    //                    if (exception!=null){
    //                        System.out.println("提交失败的偏移量："+offsets);
    //                        exception.printStackTrace();
    //
    //                    }
    //                }
    //            });//异步提交 如果提交失败错误信息会被记录下来



            }
        }catch (Exception e){
            e.printStackTrace();
        }

    }

}
